//==================================================================
//  Copyright (C) 2006-2007  Davide Pasca
//
//  This library is free software; you can redistribute it and/or
//  modify it under the terms of the GNU Lesser General Public
//  License as published by the Free Software Foundation; either
//  version 2.1 of the License, or (at your option) any later version.
//
//  This library is distributed in the hope that it will be useful,
//  but WITHOUT ANY WARRANTY; without even the implied warranty of
//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
//  Lesser General Public License for more details.
//
//  You should have received a copy of the GNU Lesser General Public
//  License along with this library; if not, write to the Free Software
//  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
//==================================================================
///
///
///
///
//==================================================================

#include "pnetlib_drtysock.h"
#include "pnetlib_compak.h"
#include "pnetlib_socksupport.h"

#pragma comment ( lib, "wsock32.lib" )

//==================================================================
#define PRINT_SOCK_ERR
static u_int	_ticket_cnt=1;	// 0 is reserved for no ticket

#define DEBUG_TRAFFIC

//==================================================================
static int setup_nodelay_noblock( int fd )
{
	unsigned long int	on = 0;
	u_long				blockflg;

	blockflg = 1;
	if ( ioctlsocket( fd, FIONBIO, &blockflg ) < 0 )
	{
		PRINT_SOCK_ERR;
		return -1;
	}

	return 0;
}

//=============================================================================
///
//=============================================================================
//namespace PASCALIBS {

//==================================================================
///
//==================================================================
ComListener::ComListener() :
	_listen_fd(-1)
{
}

//==================================================================
ComListener::~ComListener()
{
	StopListen();
}

//==================================================================
int ComListener::StartListen( int port_number )
{
	struct sockaddr_in	in_sa;
	u_long				blockflg;

	StopListen();
	_listen_fd = socket(PF_INET, SOCK_STREAM, 0);
	if ( _listen_fd < 0 )
		return COM_ERR_GENERIC;

	memset( &in_sa, 0, sizeof(in_sa) );
	in_sa.sin_family = AF_INET;
	in_sa.sin_addr.s_addr = htonl( INADDR_ANY );
	in_sa.sin_port        = htons( port_number );
	if ( bind( _listen_fd, (struct sockaddr *)&in_sa, sizeof(in_sa) ) == -1 ||
		listen( _listen_fd, SOMAXCONN )                              == -1 )
	{
		StopListen();
		return COM_ERR_GENERIC;
	}

	blockflg = 1;
	if ( ioctlsocket( _listen_fd, FIONBIO, &blockflg ) < 0 )
	{
		StopListen();
		return COM_ERR_GENERIC;
	}

	return COM_ERR_NONE;
}

//==================================================================
void ComListener::StopListen()
{
	if ( _listen_fd >= 0 )
	{
		closesocket( _listen_fd );
		_listen_fd = -1;
	}

	_listen_port = 0;
}

//==================================================================
int ComListener::Idle( int &out_accepted_fd )
{
	struct timeval	tv;
	fd_set			rdset;

	out_accepted_fd = -1;

	if ( _listen_fd >= 0 )
	{
		FD_ZERO( &rdset );
		FD_SET( _listen_fd, &rdset );
		tv.tv_sec = 0;
		tv.tv_usec = 0;

		if ( select( 1, &rdset, NULL, 0, &tv ) == -1 )
		{
			StopListen();
			return COM_ERR_GENERIC;
		}

		if ( FD_ISSET( _listen_fd, &rdset ) )
		{
			out_accepted_fd = accept( _listen_fd, 0, 0 );
			if ( out_accepted_fd >= 0 )
			{
				if ( setup_nodelay_noblock( out_accepted_fd ) )
				{
					closesocket( out_accepted_fd );
					out_accepted_fd = -1;
					return COM_ERR_GENERIC;
				}

				return COM_ERR_CONNECTED;
			}
		}
	}

	return COM_ERR_NONE;
}

//==================================================================
///
//==================================================================
typedef enum
{
	NOT_CONNECTED=0,
	CONNECTING=1,
	CONNECTED=2
} status_e;

enum {
	TH_STATUS_NULL,
	TH_STATUS_RUN,
	TH_STATUS_REQ_QUIT,
	TH_STATUS_REQ_RECYCLE,	// makes sure that the thread is staring a fresh look with the new FD
	TH_STATUS_DID_QUIT
};

//==================================================================
Compak::Compak() :
	_status(NOT_CONNECTED),
	_connected_as_caller(false),
	_fd(-1),
	_top_pack_done_bytes(0),
	_io_mutex_h(NULL),
	_hservice(NULL),
	_thread_status(TH_STATUS_NULL),
	_need_disconnect(false),
	_on_pack_pkid(0),
	_on_pack_userdatap(NULL),
	_on_pack_callback(NULL),
	_connection_started_time(0),
	_cur_inpack_readbytes(0)
{
	DWORD	thread_id;

	_io_mutex_h = CreateMutex( NULL, 0, _T( "outpack mutex" ) );
	//InitializeCriticalSection( &cp->_cri_section );
	_hservice = (void *)CreateThread( NULL, 0, (LPTHREAD_START_ROUTINE)Compak::ioThread_s, this, 0, &thread_id );
	_thread_status = TH_STATUS_RUN;
}

//==================================================================
void CompakStats::Reset()
{
	_total_send_bytes = 0;
	_total_recv_bytes = 0;
	_start_timed = 0;

	_send_bytes_queue = 0;
	_recv_bytes_queue = 0;

	_total_send_bytes_window = 0;
	_total_recv_bytes_window = 0;
	_start_timed_window = 0;
	_window_size = 0;

	_send_bytes_per_sec_window = 0;
	_recv_bytes_per_sec_window = 0;


	double timed = PSYS::TimerGetD();
	_start_timed = timed;
	_start_timed_window = timed;
	_window_size = 5 * 1000;
}

//==================================================================
void CompakStats::UpdateWindow( double timed )
{
double	delta;

	delta = timed - _start_timed_window;
	if ( delta >= _window_size )
	{
		delta = 1000 / delta;
		_send_bytes_per_sec_window = (float)(_total_send_bytes_window * delta);
		_recv_bytes_per_sec_window = (float)(_total_recv_bytes_window * delta);
		_total_send_bytes_window = 0;
		_total_recv_bytes_window = 0;
		_start_timed_window = timed;
	}
}

/*
//==================================================================
static void quit_thread( Compak *T )
{
	//EnterCriticalSection( &T->_cri_section );
		T->_thread_status = TH_STATUS_REQ_QUIT;
	//LeaveCriticalSection( &T->_cri_section );
	while ( T->_thread_status != TH_STATUS_DID_QUIT )
	{
		Sleep( 10 );
	}
}
*/

//==================================================================
//=
//==================================================================
Compak::~Compak()
{
	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		if ( _fd >= 0 )
			closesocket( _fd );

		resetIOActivity();

		_fd = -1;

		if ( _thread_status == TH_STATUS_RUN )
		{
			_thread_status = TH_STATUS_REQ_QUIT;
			ReleaseMutex( _io_mutex_h );
		}
		else
		{
			ReleaseMutex( _io_mutex_h );
			return;
		}
	}

	do {
		Sleep( 10 );
		if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
		{
			int state = _thread_status;
			ReleaseMutex( _io_mutex_h );

			if ( state == TH_STATUS_DID_QUIT )
				break;
		}
	} while(true);
}

//==================================================================
void Compak::resetIOActivity()
{
	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		_need_disconnect = false;

		for (int i=0; i < _outpack_queue.size(); ++i)
			_outpack_queue[i].Free();
		
		_outpack_queue.clear();

		for (int i=0; i < _inpack_queue.size(); ++i)
			_inpack_queue[i].Free();

		_inpack_queue.clear();

		_top_pack_done_bytes = 0;

		ReleaseMutex( _io_mutex_h );
	}
	else
	{
		PASSERT( 0 );
	}
}

//==================================================================
void Compak::onConnect()
{
	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		setStatus( CONNECTED );
		resetIOActivity();

		ReleaseMutex( _io_mutex_h );
	}
}

//==================================================================
int Compak::onDisconnect( int printerror )
{
	int	err = LAST_SOCK_ERR;
	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		//PASSERT( _status != NOT_CONNECTED );

		setStatus( NOT_CONNECTED );

		resetIOActivity();

		closesocket( _fd );
		_fd = -1;

		ReleaseMutex( _io_mutex_h );
	}

	if ( err == ECONNRESET )
		return COM_ERR_HARD_DISCONNECT;
	else
	if ( err == EINVAL )
		return COM_ERR_INVALID_ADDRESS;
	//ECONNABORTED

	return COM_ERR_GRACEFUL_DISCONNECT;
}

//==================================================================
int Compak::Call( const TCHAR *ipnamep, int port_number )
{
	struct hostent		*hp;
	u_long				addr;
	struct sockaddr_in	out_sa;

	char *ipname_ansip = PSYS::TCHARToANSI( ipnamep );

	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		if ( _status != NOT_CONNECTED )
		{
			int	retval = COM_ERR_ALREADY_CONNECTED;
			ReleaseMutex( _io_mutex_h );
			delete [] ipname_ansip;
			return retval;
		}

		setStatus( CONNECTING );
		_connection_started_time = PSYS::TimerGetD();
		ReleaseMutex( _io_mutex_h );
	}

	addr = inet_addr( ipname_ansip );
	if ( addr == INADDR_NONE )
	{
		if NOT( hp = gethostbyname( ipname_ansip ) )
		{
			if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
			{
				setStatus( NOT_CONNECTED );
				ReleaseMutex( _io_mutex_h );
			}
			delete [] ipname_ansip;
			return COM_ERR_INVALID_ADDRESS;
		}
		addr = *(long *)hp->h_addr;
	}
	delete [] ipname_ansip;

	memset( &out_sa, 0, sizeof(out_sa) );
	memcpy( &out_sa.sin_addr, &addr, sizeof(long) );
	out_sa.sin_family = AF_INET;
	out_sa.sin_port = htons( port_number );


	int	fd;

	if ( (fd = socket(AF_INET, SOCK_STREAM, 0)) < 0 )
	{
		PRINT_SOCK_ERR;

		if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
		{
			setStatus( NOT_CONNECTED );
			ReleaseMutex( _io_mutex_h );
		}
		return -1;
	}

	if ( setup_nodelay_noblock( fd ) )
	{
		PRINT_SOCK_ERR;
		closesocket( fd );

		if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
		{
			setStatus( NOT_CONNECTED );
			ReleaseMutex( _io_mutex_h );
		}

		return -1;
	}

	//-----------------------------------
	if ( connect( fd, (struct sockaddr *)&out_sa, sizeof(out_sa)) < 0 )
	{
		int	err = LAST_SOCK_ERR;

		if ( err )
		{
			if ( err != EISCONN && err != EWOULDBLOCK && err != EALREADY )
			{
				closesocket( fd );

				if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
				{
					setStatus( NOT_CONNECTED );
					ReleaseMutex( _io_mutex_h );
				}

				if ( err == ECONNRESET )
					return COM_ERR_HARD_DISCONNECT;
				else
				if ( err == EINVAL )
					return COM_ERR_INVALID_ADDRESS;

				return COM_ERR_GENERIC;
			}
		}
	}

	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		_fd = fd;
		ReleaseMutex( _io_mutex_h );
	}

	return COM_ERR_NONE;
}

/*
if ( err == EINVAL )
{
	printf( "EINVAL !!! %i\n", cp->_einval_cnt );
	if ( ++cp->_einval_cnt < 100 )
	{
		Sleep(5);
		return COM_ERR_NONE;
	}
	else
		return onDisconnect( 0 );
}
else
	return onDisconnect( 1 );
*/

//==================================================================
int Compak::SetAcceptedConn( int fd )
{
	int	ret_val;

	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		if ( _status != NOT_CONNECTED )
		{
			ret_val = COM_ERR_ALREADY_CONNECTED;
			ReleaseMutex( _io_mutex_h );
			return ret_val;
		}

		_fd = fd;
		_connected_as_caller = false;
		setStatus( CONNECTED );
		_stats.Reset();

		ReleaseMutex( _io_mutex_h );
	}

	return COM_ERR_NONE;
}

//==================================================================
void Compak::Disconnect()
{
	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
//		PASSERT( _status != NOT_CONNECTED );
		setStatus( NOT_CONNECTED );

		if ( _fd >= 0 )
			onDisconnect( false );

		resetIOActivity();

		if ( _fd >= 0 )
		{
			closesocket( _fd );
			_fd = -1;
		}

		ReleaseMutex( _io_mutex_h );
	}
}


//==================================================================
static int err_send( int fd, const u_char *datap, int size, int *sent_lenp )
{
	*sent_lenp = send( fd, (const char *)datap, size, 0 );
	if ( *sent_lenp <= 0 )
	{
	int	err;
		switch ( err = LAST_SOCK_ERR )
		{
		case EWOULDBLOCK:
			if ( *sent_lenp == 0 )
				return COM_ERR_GRACEFUL_DISCONNECT;

			return COM_ERR_NONE;

		case ECONNRESET:
			return COM_ERR_HARD_DISCONNECT;

		default:
			return COM_ERR_HARD_DISCONNECT;
		}
	}

	return COM_ERR_NONE;
}

//==================================================================
int Compak::handleSend()
{
	int		sent_len, err;

	for (int i=0; i < _outpack_queue.size(); ++i)
	{
		CompakOutpack	*op = &_outpack_queue[i];

		err = err_send( _fd, (u_char *)op->_head_and_datap + _top_pack_done_bytes,
						op->_total_bytes - _top_pack_done_bytes,
						&sent_len );
		if ( err )
			return err;

		if ( sent_len <= 0 )		// couldn't send ? would block !
			return COM_ERR_NONE;

		_top_pack_done_bytes += sent_len;

		if ( _top_pack_done_bytes >= op->_total_bytes )
		{
			_top_pack_done_bytes = 0;

			op->Free();
			_outpack_queue.erase( _outpack_queue.begin() + i );
			--i;
		}
		else
			break;
	}
	
	return COM_ERR_NONE;
}


//==================================================================
static int err_recv( int fd, u_char *datap, int size, int *recvd_lenp )
{
	*recvd_lenp = recv( fd, (char *)datap, size, 0 );
	if ( *recvd_lenp <= 0 )
	{
		switch ( LAST_SOCK_ERR )
		{
		case EWOULDBLOCK:
			if ( *recvd_lenp == 0 )
				return COM_ERR_GRACEFUL_DISCONNECT;

			return COM_ERR_NONE;

		case ECONNRESET:
			return COM_ERR_HARD_DISCONNECT;

		default:
			return COM_ERR_HARD_DISCONNECT;
		}
	}

	return COM_ERR_NONE;
}

//==================================================================
int Compak::handleReceive()
{
	int	recvd_len, err;

	while (true)
	{
		while ( _cur_inpack_readbytes < sizeof(packet_header_t) )
		{
			err = err_recv( _fd, (u_char *)&_cur_inpack_head + _cur_inpack_readbytes,
							sizeof(packet_header_t) - _cur_inpack_readbytes,
							&recvd_len );
			if ( err )
				return err;

			if ( recvd_len <= 0 )		// couldn't receive ? would block
				return COM_ERR_NONE;

			_cur_inpack_readbytes += recvd_len;
			PASSERT( _cur_inpack_readbytes <= sizeof(packet_header_t) );

			// if head is done, then
			if ( _cur_inpack_readbytes == sizeof(packet_header_t) )
			{
				// copy head and allocate necessary data
				_inpack_queue.resize( _inpack_queue.size()+1 );

				_inpack_queue.back().Init( _cur_inpack_head );
			}

			//----- update recv stats
			_stats._total_recv_bytes += recvd_len;
			_stats._total_recv_bytes_window += recvd_len;
		}

		if ( _inpack_queue.size() )
		{
			CompakInpack	*ip = &_inpack_queue[ _inpack_queue.size()-1 ];

			u_int	data_readbytes = _cur_inpack_readbytes - sizeof(packet_header_t);

			while ( data_readbytes < ip->GetDataSize() )
			{
				err = err_recv( _fd,
								(u_char *)ip->GetData() + data_readbytes,
								ip->GetDataSize() - data_readbytes,
								&recvd_len );
				if ( err )
					return err;

				if ( recvd_len <= 0 )		// couldn't receive ? would block
					return COM_ERR_NONE;

				_cur_inpack_readbytes += recvd_len;
				data_readbytes = _cur_inpack_readbytes - sizeof(packet_header_t);

				//----- update recv stats
				_stats._total_recv_bytes += recvd_len;
				_stats._total_recv_bytes_window += recvd_len;
			}

			// packet completely received !!
			ip->_is_done = true;

			// reset the current packet read bytes in input
			_cur_inpack_readbytes = 0;

			if ( ip->_head_and_datap->id == _on_pack_pkid && _on_pack_callback )
			{
				_on_pack_callback( ip->GetData(), ip->GetDataSize(), _on_pack_userdatap );
				disposeInPack( ip );
			}
		}
	}

	return COM_ERR_NONE;
}

//==================================================================
DWORD WINAPI Compak::ioThread_s( Compak *mythis )
{
	return mythis->ioThread();
}

//==================================================================
DWORD Compak::ioThread()
{
struct timeval	tv;
fd_set			rdset;
fd_set			wtset;
int				err;

	while (true)
	{
		u_int	sleep_ms = 5;

		if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
		{
			if ( _thread_status == TH_STATUS_REQ_QUIT )
			{
				_thread_status = TH_STATUS_DID_QUIT;
				ReleaseMutex( _io_mutex_h );
				return 0;
			}
			else
			if ( _thread_status == TH_STATUS_RUN )
			{
				if ( _status == CONNECTED )
				{
					FD_ZERO( &rdset );
					FD_ZERO( &wtset );
					FD_SET( _fd, &rdset );

					if ( _outpack_queue.size() )
						FD_SET( _fd, &wtset );

					tv.tv_sec = 0;
					tv.tv_usec = 1000;	// 10^6(=1sec) -> 10^6 / 10^4 = 10^2 (1/100th of a second)

					ReleaseMutex( _io_mutex_h );

					int	sel_result = select( 2, &rdset, &wtset, 0, &tv );

					if ( WaitForSingleObject( _io_mutex_h, INFINITE ) != WAIT_OBJECT_0 )
					{
						PASSERT( 0 );
						return -1;
					}

					if ( sel_result != -1 )
					{
						if ( FD_ISSET( _fd, &wtset ) || FD_ISSET( _fd, &rdset ) )
						{
							//if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
							{
								if ( FD_ISSET( _fd, &rdset ) )
								{
									if ( err = handleReceive() )
									{
										_need_disconnect = true;
									}
								}

								if ( FD_ISSET( _fd, &wtset ) )
								{
									if ( err = handleSend() )
									{
										_need_disconnect = true;
									}
								}

							}
						}
					}
				}
				else
				{
					sleep_ms = 100;
				}
			}

			ReleaseMutex( _io_mutex_h );
			Sleep( sleep_ms );
		}
	}
/*
	if ( _thread_status == TH_STATUS_REQ_RECYCLE )
	{
		_thread_status = TH_STATUS_RUN;
		goto recycle;
	}
	else
		_thread_status = TH_STATUS_DID_QUIT;
*/

	return 0;
}

//==================================================================
int Compak::Idle()
{
	struct timeval	tv;
	fd_set			rdset, wtset;
	double			timed;

	FD_ZERO( &rdset );
	FD_ZERO( &wtset );

	switch ( _status )
	{
	case NOT_CONNECTED:
		break;

	case CONNECTING:
		if ( (PSYS::TimerGetD() - _connection_started_time) >= 20*1000 )
		{
			onDisconnect( false );
			return COM_ERR_TIMEOUT_CONNECTING;
		}
		else
		{
			FD_SET( _fd, &wtset );
			tv.tv_sec = 0;
			tv.tv_usec = 0;
			if ( select( 1, &rdset, &wtset, 0, &tv ) == -1 )
				return onDisconnect( 1 );
	
			if ( FD_ISSET( _fd, &wtset ) )
			{					
				//---- reset the stats on a new connection
				_stats.Reset();
	
				_connected_as_caller = true;
				onConnect();
				return COM_ERR_CONNECTED;
			}
		}
		break;

	case CONNECTED:
		if ( _need_disconnect )
		{
			onDisconnect( 1 );

			_need_disconnect = false;
			return COM_ERR_GENERIC;
		}

		break;
	}

	timed = PSYS::TimerGetD();
	_stats.UpdateWindow( timed );

	return COM_ERR_NONE;
}

//==================================================================
bool Compak::IsConnected() const
{
	return _status == CONNECTED;
}

//==================================================================
bool Compak::GetInputPack( u_int *out_pack_idp,
							 u_int *out_pack_data_sizep,
							 void *out_datap,
							 u_int out_data_max_size,
							 int n_pack_filter,
							 const u_int *filter_id_array )
{
	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		if NOT( _inpack_queue.size() )
		{
			ReleaseMutex( _io_mutex_h );
			return false;
		}

		CompakInpack	*ip = &_inpack_queue[0];

		if ( ip->_head_and_datap && ip->_is_done )
		{
			if ( n_pack_filter > 0 )
			{
				int	i;
				for (i=0; i < n_pack_filter; ++i)
					if ( filter_id_array[i] == ip->_head_and_datap->id )
						break;
			
				if ( i == n_pack_filter )
					return false;
			}

			PASSERT( ip->GetDataSize() <= out_data_max_size );

			*out_pack_idp = ip->_head_and_datap->id;
			*out_pack_data_sizep = ip->GetDataSize();

			memcpy( out_datap, ip->GetData(), ip->GetDataSize() );

			ReleaseMutex( _io_mutex_h );
			return true;
		}
		else
		{
			ReleaseMutex( _io_mutex_h );
			return false;
		}

	}
	else
	{
		PASSERT( 0 );
		return false;
	}
}

//==================================================================
void Compak::disposeInPack( CompakInpack *ip )
{
	ip->Free();
	_inpack_queue.erase( _inpack_queue.begin() + (ip - &_inpack_queue.front()) );
}

//==================================================================
void Compak::DisposeINPacket()
{
	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		if ( _inpack_queue.size() )
			disposeInPack( &_inpack_queue[0] );

		ReleaseMutex( _io_mutex_h );
	}
	else
	{
		PASSERT( 0 );
	}
}

//==================================================================
int Compak::SendPacket( u_short pk_id, const void *datap, int data_size, u_int *send_ticketp )
{
	if ( datap == NULL || data_size == 0 )
	{
		datap = NULL;
		data_size = 0;
	}

	void	*packp = AllocPacket( pk_id, data_size );
	if NOT( packp )
	{
		PASSERT( 0 );
		return -1;
	}

	if ( datap )
		memcpy( packp, datap, data_size );

	return SendPacket( packp, send_ticketp );
}

//==================================================================
int Compak::SendPacket( void *compak_allocated_datap, u_int *send_ticketp )
{
	//KBASE_DEBUG_PRINTF( "n out packs: %i\n", _outpack_queue.size() );

	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		_outpack_queue.resize( _outpack_queue.size()+1 );

		CompakOutpack	&op = _outpack_queue.back();

		packet_header_t *packetp = ((packet_header_t *)compak_allocated_datap - 1);

		op._ticket_num		= _ticket_cnt;
		op._total_bytes	= sizeof(packet_header_t) + packetp->size;
		op._head_and_datap	= (packet_header_t *)packetp;

		_stats._send_bytes_queue += op._total_bytes;
		
		if ( send_ticketp )
			*send_ticketp = _ticket_cnt;

		ReleaseMutex( _io_mutex_h );
	}
	else
	{
		PASSERT( 0 );
	}

	++_ticket_cnt;	
	if ( _ticket_cnt == 0 )	// 0 is reserved for no ticket
		_ticket_cnt = 1;

	return COM_ERR_NONE;
}

//==================================================================
void *Compak::AllocPacket( u_short pk_id, int data_size )
{
	packet_header_t	*p;

	// alloc the memory for the copy of the data buffer to transmit
	if NOT( p = (packet_header_t *)malloc( sizeof(packet_header_t) + data_size ) )
		return NULL;

	p->id	= pk_id;
	p->pad	= 0;
	p->size	= data_size;

	return (void *)(p + 1);
}

//==================================================================
int Compak::SearchOUTQueue( u_short packet_id )
{
	int		cnt = 0;

	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		for (int i=0; i < _outpack_queue.size(); ++i)
		{
			if ( _outpack_queue[i]._head_and_datap->id == packet_id )
				++cnt;
		}
		ReleaseMutex( _io_mutex_h );
	}
	else
	{
		PASSERT( 0 );
	}

	return cnt;
}

//==================================================================
int Compak::GetOUTQueueCnt() const
{
	int		cnt;

	if ( WaitForSingleObject( _io_mutex_h, INFINITE ) == WAIT_OBJECT_0 )
	{
		cnt = _outpack_queue.size();
		ReleaseMutex( _io_mutex_h );
	}
	else
	{
		PASSERT( 0 );
	}

	return cnt;
}

//==================================================================
void Compak::setStatus( int status )
{
	_status = status;
	if ( _status == 0 )
	{
		_status = 0;
	}
}

//==================================================================
/*
int compak_is_data_sent( Compak *cp, u_int send_ticket )
{
	if ( cp->_outdata_ticket == send_ticket )
		return 0;

	return 1;
}
*/

//=============================================================================
//}